package com.wudl.flink.stream.sink

import java.sql.{Connection, DriverManager, PreparedStatement}

import com.wudl.flink.stream.source.TransformDemo.SensorReading
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction
import org.apache.flink.streaming.api.functions.source.SourceFunction
import org.apache.flink.streaming.api.scala._

import scala.util.Random


object MysqlSink {
  def main(args: Array[String]): Unit = {


    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    val inputStream = env.readTextFile("F:\\ideaWorkSpace2020\\demo\\Flink-wudl\\src\\main\\resources\\sensor.txt")

    val stream = env.addSource(new MySensorSource())
    val streamSink = stream.addSink(new MyJdbcSinkFun())

    env.execute("MysqlSink")
  }

  class MyJdbcSinkFun() extends RichSinkFunction[SensorReading] {

    // 定义预编译语句
    var conn: Connection = _
    var insertStmt: PreparedStatement = _
    var updateStmt: PreparedStatement = _

    override def invoke(value: SensorReading): Unit = {
      updateStmt.setDouble(1, value.temperature)
      updateStmt.setString(2, value.id)
      updateStmt.execute()
      // 如果没有更新到就插入
      if (updateStmt.getUpdateCount == 0) {
        insertStmt.setString(1, value.id)
        insertStmt.setDouble(2, value.temperature)
        insertStmt.execute()
      }
    }

    override def open(parameters: Configuration): Unit = {

      conn = DriverManager.getConnection("jdbc:mysql://10.204.125.109:3306/test", "root", "123456")
      insertStmt = conn.prepareStatement("insert into testdemo (id,name) values (?,?)")
      updateStmt = conn.prepareStatement("update testdemo set name = ? where id = ? ")
    }

    override def close(): Unit = {
      insertStmt.close()
      updateStmt.close()
      conn.close()

    }
  }

  // 自定义SourceFunction
  class MySensorSource() extends SourceFunction[SensorReading] {
    // 定义一个标识位flag，用来表示数据源是否正常运行发出数据
    var running: Boolean = true

    override def cancel(): Unit = running = false

    override def run(ctx: SourceFunction.SourceContext[SensorReading]): Unit = {
      // 定义一个随机数发生器
      val rand = new Random()

      // 随机生成一组（10个）传感器的初始温度: （id，temp）
      var curTemp = 1.to(10).map(i => ("sensor_" + i, rand.nextDouble() * 100))

      // 定义无限循环，不停地产生数据，除非被cancel
      while (running) {
        // 在上次数据基础上微调，更新温度值
        curTemp = curTemp.map(
          data => (data._1, data._2 + rand.nextGaussian())
        )
        // 获取当前时间戳，加入到数据中，调用ctx.collect发出数据
        val curTime = System.currentTimeMillis()
        curTemp.foreach(
          data => ctx.collect(SensorReading(data._1, curTime, data._2))
        )
        // 间隔500ms
        Thread.sleep(500)
      }
    }
  }
}
